#include "protocol/mqtt/MqttClient.h"
#include "core/Logger.h"

#include <mosquitto.h>
#include <chrono>
#include <algorithm>
#include <string>
#include <cctype>
#include <cstdio>

namespace ems {
namespace protocol {

namespace {
// 将紧凑JSON字符串格式化为缩进良好的多行文本；非JSON则原样返回
static std::string prettyJson(const std::string& src) {
    auto it = std::find_if(src.begin(), src.end(), [](unsigned char ch){ return !std::isspace(ch); });
    if (it == src.end() || (*it != '{' && *it != '[')) {
        return src; // 非JSON，直接返回
    }

    std::string out;
    out.reserve(src.size() + src.size() / 4);
    const int indentSize = 2;
    int depth = 0;
    bool inString = false;
    bool escape = false;

    for (size_t i = 0; i < src.size(); ++i) {
        unsigned char ch = static_cast<unsigned char>(src[i]);

        if (escape) { out.push_back(ch); escape = false; continue; }
        if (ch == '\\') { out.push_back(ch); if (inString) escape = true; continue; }
        if (ch == '"') { out.push_back(ch); inString = !inString; continue; }
        if (inString) { out.push_back(ch); continue; }

        switch (ch) {
            case '{': case '[':
                out.push_back(ch); out.push_back('\n'); depth++; out.append(depth * indentSize, ' '); break;
            case '}': case ']':
                out.push_back('\n'); depth = std::max(0, depth - 1); out.append(depth * indentSize, ' '); out.push_back(ch); break;
            case ',':
                out.push_back(ch); out.push_back('\n'); out.append(depth * indentSize, ' '); break;
            case ':':
                out.append(": "); break;
            default:
                if (!std::isspace(ch)) out.push_back(ch);
                break;
        }
    }
    return out;
}
static std::string previewPayload(const std::string& payload, size_t maxBytes = 2048) {
    std::string pretty = prettyJson(payload);
    if (pretty.size() <= maxBytes) return pretty;
    std::string out = pretty.substr(0, maxBytes);
    out += "...(truncated)";
    return out;
}
}

MqttClient::MqttClient()
    : client_(nullptr), port_(1883), keepAlive_(60), cleanSession_(true),
      autoReconnect_(true), reconnectDelay_(5000), connected_(false), loopRunning_(false) {
    
    mosquitto_lib_init();
}

MqttClient::~MqttClient() {
    stopLoop();
    disconnect();
    
    if (client_) {
        mosquitto_destroy(client_);
        client_ = nullptr;
    }
    
    mosquitto_lib_cleanup();
}

bool MqttClient::initialize(const std::string& clientId, const std::string& host, int port) {
    clientId_ = clientId;
    host_ = host;
    port_ = port;
    
    if (client_) {
        mosquitto_destroy(client_);
    }
    
    client_ = mosquitto_new(clientId_.c_str(), cleanSession_, this);
    if (!client_) {
        setLastError("Failed to create mosquitto client");
        LOG_MQTT_ERROR("MQTT init failed: create client");
        return false;
    }
    
    mosquitto_connect_callback_set(client_, onConnect);
    mosquitto_disconnect_callback_set(client_, onDisconnect);
    mosquitto_message_callback_set(client_, onMessage);
    mosquitto_publish_callback_set(client_, onPublish);
    mosquitto_subscribe_callback_set(client_, onSubscribe);
    mosquitto_unsubscribe_callback_set(client_, onUnsubscribe);
    mosquitto_log_callback_set(client_, onLog);
    
    LOG_INFO("MQTT init ok: " << clientId_ << " -> " << host_ << ":" << port_);
    return true;
}

bool MqttClient::setCredentials(const std::string& username, const std::string& password) {
    username_ = username;
    password_ = password;
    
    if (client_) {
        int result = mosquitto_username_pw_set(client_, 
            username_.empty() ? nullptr : username_.c_str(),
            password_.empty() ? nullptr : password_.c_str());
        if (result != MOSQ_ERR_SUCCESS) {
            setLastError("Failed to set credentials: " + std::string(mosquitto_strerror(result)));
            LOG_MQTT_ERROR("MQTT auth set failed: " << mosquitto_strerror(result));
            return false;
        }
        LOG_INFO("MQTT auth set ok (username only logged): " << (username_.empty()?"<empty>":username_));
    }
    
    return true;
}

bool MqttClient::setTLS(const std::string& caFile, const std::string& certFile, const std::string& keyFile) {
    caFile_ = caFile;
    certFile_ = certFile;
    keyFile_ = keyFile;
    
    if (client_ && !caFile_.empty()) {
        int result = mosquitto_tls_set(client_, 
            caFile_.c_str(),
            nullptr,  // capath
            certFile_.empty() ? nullptr : certFile_.c_str(),
            keyFile_.empty() ? nullptr : keyFile_.c_str(),
            nullptr); // password callback
            
        if (result != MOSQ_ERR_SUCCESS) {
            setLastError("Failed to set TLS: " + std::string(mosquitto_strerror(result)));
            LOG_MQTT_ERROR("MQTT TLS set failed: " << mosquitto_strerror(result));
            return false;
        }
        
        mosquitto_tls_opts_set(client_, 1, "tlsv1.2", nullptr);
        LOG_INFO("MQTT TLS set ok (ca=" << (caFile_.empty()?"<none>":caFile_) << ", cert="
                 << (certFile_.empty()?"<none>":certFile_) << ", key="
                 << (keyFile_.empty()?"<none>":keyFile_) << ")");
    }
    
    return true;
}

bool MqttClient::connect() {
    if (!client_) {
        setLastError("Client not initialized");
        return false;
    }
    
    if (connected_) {
        return true;
    }
    
    LOG_MQTT_CONN("MQTT connecting to " << host_ << ":" << port_ << ", keepAlive=" << keepAlive_
             << ", cleanSession=" << (cleanSession_?"true":"false"));
    int result = mosquitto_connect(client_, host_.c_str(), port_, keepAlive_);
    if (result != MOSQ_ERR_SUCCESS) {
        setLastError("Failed to connect: " + std::string(mosquitto_strerror(result)));
        LOG_MQTT_ERROR("MQTT connect() call failed: " << mosquitto_strerror(result));
        return false;
    }
    
    auto start = std::chrono::steady_clock::now();
    auto timeout = std::chrono::seconds(30);
    
    while (!connected_ && (std::chrono::steady_clock::now() - start) < timeout) {
        mosquitto_loop(client_, 100, 1);
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
    
    if (!connected_) {
        setLastError("Connection timeout");
        LOG_MQTT_ERROR("MQTT connect timeout waiting for CONNACK");
        return false;
    }
    
    LOG_MQTT_CONN("MQTT connect ok (CONNACK)");
    return true;
}

void MqttClient::disconnect() {
    if (client_ && connected_) {
        mosquitto_disconnect(client_);
        connected_ = false;
        LOG_MQTT_CONN("MQTT client disconnected");
    }
}

bool MqttClient::isConnected() const {
    return connected_;
}

bool MqttClient::subscribe(const std::string& topic, MqttQoS qos) {
    if (!client_ || !connected_) {
        setLastError("Not connected");
        return false;
    }
    
    int result = mosquitto_subscribe(client_, nullptr, topic.c_str(), static_cast<int>(qos));
    if (result != MOSQ_ERR_SUCCESS) {
        setLastError("Failed to subscribe: " + std::string(mosquitto_strerror(result)));
        LOG_MQTT_ERROR("MQTT subscribe fail: topic=" << topic << ", err=" << mosquitto_strerror(result));
        return false;
    }
    
    LOG_INFO("MQTT subscribe ok: " << topic);
    return true;
}

bool MqttClient::unsubscribe(const std::string& topic) {
    if (!client_ || !connected_) {
        setLastError("Not connected");
        return false;
    }
    
    int result = mosquitto_unsubscribe(client_, nullptr, topic.c_str());
    if (result != MOSQ_ERR_SUCCESS) {
        setLastError("Failed to unsubscribe: " + std::string(mosquitto_strerror(result)));
        LOG_MQTT_ERROR("MQTT unsubscribe fail: topic=" << topic << ", err=" << mosquitto_strerror(result));
        return false;
    }
    
    LOG_INFO("MQTT unsubscribe ok: " << topic);
    return true;
}

bool MqttClient::publish(const std::string& topic, const std::string& payload, MqttQoS qos, bool retain) {
    if (!client_ || !connected_) {
        setLastError("Not connected");
        return false;
    }
    
    int result = mosquitto_publish(client_, nullptr, topic.c_str(),
                                  payload.length(), payload.c_str(),
                                  static_cast<int>(qos), retain);
    std::string preview = previewPayload(payload);
    if (result != MOSQ_ERR_SUCCESS) {
        setLastError("Failed to publish: " + std::string(mosquitto_strerror(result)));
        LOG_MQTT_ERROR("MQTT publish fail: topic=" << topic << ", bytes=" << payload.length()
                  << ", qos=" << static_cast<int>(qos) << ", retain=" << (retain?"true":"false")
                  << ", err=" << mosquitto_strerror(result) << ", payload=\n" << preview);
        return false;
    }
    
    LOG_MQTT_SEND("MQTT publish ok: topic=" << topic << ", bytes=" << payload.length()
                  << ", qos=" << static_cast<int>(qos) << ", retain=" << (retain?"true":"false")
                  << ", payload=\n" << preview);
    return true;
}

bool MqttClient::publishAsync(const std::string& topic, const std::string& payload, MqttQoS qos, bool retain) {
    // 入队前打印待发送的异步消息，确保所有发送的MQTT消息都有日志
    {
        std::string preview = previewPayload(payload);
        LOG_MQTT_SEND("MQTT enqueue publish: topic=" << topic
                      << ", bytes=" << payload.length()
                      << ", qos=" << static_cast<int>(qos)
                      << ", retain=" << (retain?"true":"false")
                      << ", payload=\n" << preview);
    }

    {
        std::lock_guard<std::mutex> lock(publishQueueMutex_);
        publishQueue_.push(MqttMessage(topic, payload, qos, retain));
    }
    
    return true;
}

void MqttClient::startLoop() {
    if (loopRunning_) {
        return;
    }
    
    loopRunning_ = true;
    loopThread_ = std::thread(&MqttClient::loopForever, this);
    
    if (autoReconnect_) {
        reconnectThread_ = std::thread(&MqttClient::reconnectLoop, this);
    }
    
    LOG_INFO("MQTT client loop started");
}

void MqttClient::stopLoop() {
    if (!loopRunning_) {
        return;
    }
    
    loopRunning_ = false;
    
    if (loopThread_.joinable()) {
        loopThread_.join();
    }
    
    if (reconnectThread_.joinable()) {
        reconnectThread_.join();
    }
    
    LOG_INFO("MQTT client loop stopped");
}

void MqttClient::loopForever() {
    try {
        while (loopRunning_) {
            if (client_) {
                int result = mosquitto_loop(client_, 100, 1);
                if (result != MOSQ_ERR_SUCCESS) {
                    if (connected_) {
                        LOG_MQTT_ERROR("MQTT loop error: " << mosquitto_strerror(result));
                        connected_ = false;
                        if (connectCallback_) {
                            connectCallback_(false);
                        }
                    }
                }
            }
            
            processPublishQueue();
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    } catch (const std::system_error& e) {
        std::fprintf(stderr, "[MqttClient::loopForever] system_error: %s\n", e.what());
        throw;
    }
}

void MqttClient::reconnectLoop() {
    try {
        while (loopRunning_) {
            if (!connected_ && autoReconnect_) {
                LOG_INFO("Attempting to reconnect to MQTT broker...");
                
                if (client_) {
                    int result = mosquitto_reconnect(client_);
                    if (result == MOSQ_ERR_SUCCESS) {
                        LOG_MQTT_CONN("MQTT reconnect initiated");
                    } else {
                        LOG_MQTT_ERROR("MQTT reconnect failed: " << mosquitto_strerror(result));
                    }
                }
            }
            
            std::this_thread::sleep_for(std::chrono::milliseconds(reconnectDelay_));
        }
    } catch (const std::system_error& e) {
        std::fprintf(stderr, "[MqttClient::reconnectLoop] system_error: %s\n", e.what());
        throw;
    }
}

void MqttClient::processPublishQueue() {
    if (!connected_) {
        return;
    }
    
    std::lock_guard<std::mutex> lock(publishQueueMutex_);
    
    while (!publishQueue_.empty()) {
        const auto& msg = publishQueue_.front();
        
        int result = mosquitto_publish(client_, nullptr, msg.topic.c_str(),
                                      msg.payload.length(), msg.payload.c_str(),
                                      static_cast<int>(msg.qos), msg.retain);
        
        // 成功或失败都打印payload，方便排查
        std::string preview = previewPayload(msg.payload);
        if (result == MOSQ_ERR_SUCCESS) {
            LOG_MQTT_SEND("MQTT publish ok (async): topic=" << msg.topic << ", bytes=" << msg.payload.length()
                          << ", qos=" << static_cast<int>(msg.qos) << ", retain=" << (msg.retain?"true":"false")
                          << ", payload=\n" << preview);
            publishQueue_.pop();
        } else {
            LOG_MQTT_ERROR("Failed to publish queued message: topic=" << msg.topic
                           << ", bytes=" << msg.payload.length()
                           << ", qos=" << static_cast<int>(msg.qos)
                           << ", retain=" << (msg.retain?"true":"false")
                           << ", err=" << mosquitto_strerror(result)
                           << ", payload=\n" << preview);
            break;
        }
    }
}

std::string MqttClient::getLastError() const {
    std::lock_guard<std::mutex> lock(errorMutex_);
    return lastError_;
}

void MqttClient::setLastError(const std::string& error) const {
    std::lock_guard<std::mutex> lock(errorMutex_);
    lastError_ = error;
}

void MqttClient::onConnect(mosquitto* client, void* userdata, int result) {
    (void)client;
    
    MqttClient* mqttClient = static_cast<MqttClient*>(userdata);
    
    if (result == 0) {
        mqttClient->connected_ = true;
        LOG_INFO("MQTT onConnect ok (result=0)");
    } else {
        mqttClient->connected_ = false;
        LOG_MQTT_ERROR("MQTT onConnect fail: " << mosquitto_connack_string(result));
        mqttClient->setLastError("Connection failed: " + std::string(mosquitto_connack_string(result)));
    }
    
    if (mqttClient->connectCallback_) {
        mqttClient->connectCallback_(result == 0);
    }
}

void MqttClient::onDisconnect(mosquitto* client, void* userdata, int result) {
    (void)client;
    
    MqttClient* mqttClient = static_cast<MqttClient*>(userdata);
    mqttClient->connected_ = false;
    
    if (result == 0) {
        LOG_MQTT_CONN("MQTT onDisconnect ok");
    } else {
        LOG_MQTT_CONN("MQTT onDisconnect unexpected: " << mosquitto_strerror(result));
    }
    
    if (mqttClient->connectCallback_) {
        mqttClient->connectCallback_(false);
    }
}

void MqttClient::onMessage(mosquitto* client, void* userdata, const mosquitto_message* message) {
    (void)client;
    
    MqttClient* mqttClient = static_cast<MqttClient*>(userdata);
    
    if (message && message->payload && mqttClient->messageCallback_) {
        std::string topic(message->topic);
        std::string payload(static_cast<const char*>(message->payload), message->payloadlen);
        
        std::string preview = previewPayload(payload);
        LOG_MQTT_RECV("MQTT recv: topic=" << topic << ", bytes=" << payload.length() << ", payload=\n" << preview);
        mqttClient->messageCallback_(topic, payload);
    }
}

void MqttClient::onPublish(mosquitto* client, void* userdata, int messageId) {
    (void)client;
    
    MqttClient* mqttClient = static_cast<MqttClient*>(userdata);
    
    LOG_DEBUG("MQTT message published, message ID: " << messageId);
    
    if (mqttClient->publishCallback_) {
        mqttClient->publishCallback_(messageId, true);
    }
}

void MqttClient::onSubscribe(mosquitto* client, void* userdata, int messageId, int qosCount, const int* grantedQos) {
    (void)client;
    (void)userdata;
    
    LOG_DEBUG("MQTT subscription confirmed, message ID: " << messageId << ", QoS count: " << qosCount);
    
    for (int i = 0; i < qosCount; ++i) {
        LOG_DEBUG("Granted QoS " << i << ": " << grantedQos[i]);
    }
}

void MqttClient::onUnsubscribe(mosquitto* client, void* userdata, int messageId) {
    (void)client;
    (void)userdata;
    
    LOG_DEBUG("MQTT unsubscription confirmed, message ID: " << messageId);
}

void MqttClient::onLog(mosquitto* client, void* userdata, int level, const char* str) {
    (void)client;
    (void)userdata;
    
    switch (level) {
        case MOSQ_LOG_DEBUG:
            LOG_DEBUG("MQTT: " << str);
            break;
        case MOSQ_LOG_INFO:
            LOG_INFO("MQTT: " << str);
            break;
        case MOSQ_LOG_NOTICE:
            LOG_INFO("MQTT: " << str);
            break;
        case MOSQ_LOG_WARNING:
            LOG_WARN("MQTT: " << str);
            break;
        case MOSQ_LOG_ERR:
            LOG_MQTT_ERROR("MQTT: " << str);
            break;
        default:
            LOG_INFO("MQTT: " << str);
            break;
    }
}

}  // namespace protocol
}  // namespace ems
